package com.sunseagear.wind.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Schema;
import org.apache.kudu.Type;
import org.apache.kudu.client.*;
import com.sunseagear.common.utils.entity.Principal;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * TODO
 *
 * @author KY
 * @version 1.0
 * @date 2021/4/6 20:20
 */
@Slf4j
@Service
public class KuduService {

    private KuduClient kuduClient;

    @Value("${kudu.kudu-address:10.89.3.131:7050}")
    private String kuduMaster;
    @Value("${kudu.tablename:principal}")
    private String tableName;


    public void init(){
//        kuduMaster = "10.89.3.131:7050";
//        tableName = "principal";
        KuduClient.KuduClientBuilder kuduClientBuilder = new KuduClient.KuduClientBuilder(kuduMaster);
        kuduClientBuilder.defaultSocketReadTimeoutMs(10000);
        kuduClient = kuduClientBuilder.build();
    }

    public void testCreateTable() throws KuduException {

        //        private String id; // 编号
//        private String username; // 登录名
//        private String realname; // 姓名
//        private String tenantId; //租户ID
//        private String roleId; //角色ID

        if(!kuduClient.tableExists(tableName)){
            ArrayList<ColumnSchema> columnSchemas = new ArrayList<>();
            columnSchemas.add(new ColumnSchema.ColumnSchemaBuilder("id", Type.INT32).key(true).build());
            columnSchemas.add(new ColumnSchema.ColumnSchemaBuilder("username",Type.STRING).build());
            columnSchemas.add(new ColumnSchema.ColumnSchemaBuilder("realname",Type.STRING).build());
            columnSchemas.add(new ColumnSchema.ColumnSchemaBuilder("tenantId",Type.STRING).build());
            columnSchemas.add(new ColumnSchema.ColumnSchemaBuilder("roleId",Type.STRING).build());
            columnSchemas.add(new ColumnSchema.ColumnSchemaBuilder("loginIp",Type.STRING).build());
            //日期待定
            columnSchemas.add(new ColumnSchema.ColumnSchemaBuilder("lastAccessTime",Type.UNIXTIME_MICROS).build());
            //小数待定
//            columnSchemas.add(new ColumnSchema.ColumnSchemaBuilder("money",Type.DOUBLE).build());
            Schema schema = new Schema(columnSchemas);

            CreateTableOptions options = new CreateTableOptions();
            List<String> partitionList = new ArrayList<>();
            //kudu表的分区字段是什么 TODO
            partitionList.add("id");
            //按照id.hashcode % 分区数 = 分区号
            options.addHashPartitions(partitionList,6);

            kuduClient.createTable(tableName,schema,options);
        }
    }


    public void insert() throws KuduException {
        KuduSession kuduSession = kuduClient.newSession();
        kuduSession.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);

        for(int i = 0; i <= 100;i ++){
            KuduTable userTable = kuduClient.openTable(tableName);
            Insert insert = userTable.newInsert();
            PartialRow row = insert.getRow();
            row.addInt("id",i);
            row.addString("name","wang"+i);
            row.addDouble("money",100.342+i);
            kuduSession.apply(insert);
        }
    }

    public void insert(List<Principal> list) throws KuduException {

        KuduSession kuduSession = kuduClient.newSession();
        kuduSession.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);
        for(Principal p:list){
            KuduTable userTable = kuduClient.openTable(tableName);
            Insert insert = userTable.newInsert();
            PartialRow row = insert.getRow();
            row.addInt("id",Integer.parseInt(p.getId()));
            row.addString("username",p.getUsername());
            row.addString("realname",p.getRealname());
            row.addString("tenantId",p.getTenantId());
            row.addString("roleId",p.getRoleId());
            row.addString("loginIp",p.getLoginIp());
            kuduSession.apply(insert);
        }
    }

    public void query() throws KuduException {
        KuduScanner.KuduScannerBuilder kuduScannerBuilder = kuduClient
                .newScannerBuilder(kuduClient.openTable(tableName));
        List<String> columns = Arrays.asList("id","name","moeny");
        kuduScannerBuilder.setProjectedColumnNames(columns);
        KuduScanner kuduScanner = kuduScannerBuilder.build();
        while (kuduScanner.hasMoreRows()){
            RowResultIterator rowResults = kuduScanner.nextRows();
            while(rowResults.hasNext()){
                RowResult row = rowResults.next();
                log.info("id={},name={},money={}",
                        row.getInt("id"),
                        row.getString("name"),
                        row.getDouble("money"));
            }
        }
    }

    public void update() throws KuduException {
        KuduSession kuduSession = kuduClient.newSession();
        kuduSession.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);
        KuduTable kuduTable = kuduClient.openTable(tableName);

//        Update update = kuduTable.newUpdate();
        //id存在就修改，不存在就新增
        Upsert upsert = kuduTable.newUpsert();
        PartialRow row = upsert.getRow();
        row.addInt("id",100000);
        row.addString("name","yangege");
        row.addDouble("money",100.222);
        kuduSession.apply(upsert);
    }


//    public static void main(String[] args) throws KuduException {
//        KuduService kuduService = new KuduService();
//        kuduService.init();
//        kuduService.testCreateTable();
//    }
}
